package com.example.wxg.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author void
 * @date 2021/9/3 15:33
 * @desc 需要一台linux电脑，配置本机域名devtwo，linux电脑执行 nc -l 9800，命令如果不存在就yum -install -y nc
 * 在linux输入单次，控制台会打印输出
 * 实现效果：每隔1s对最近2s内的数据进行汇总计算
 */
public class SocketWindowWordCount {

    public static void main(String[] args) throws Exception {
        int port;
        try{
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port");
        }catch (Exception e){
            e.printStackTrace();
            port = 9800;
        }
        
        //获取flink的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String hostname = "devtwo";
        String delimiter = "\n";
        //连接socket获取输入的数据
        DataStreamSource<String> text = env.socketTextStream(hostname, port, delimiter);
        
        DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {

            @Override
            public void flatMap(String s, Collector<WordWithCount> collector) throws Exception {
                String[] splits = s.split("\\s");
                for(String word : splits){
                    collector.collect(new WordWithCount(word, 1L));
                }
            }
        }).keyBy("word")
                //指定时间窗口大小为2s，指定时间间隔为1s
                .timeWindow(Time.seconds(2), Time.seconds(1)).sum("count");

        //将数据打印控制台并设置并行度
        windowCounts.print().setParallelism(1);
        env.execute("Socket window count");
    }
    
    public static class WordWithCount{
        public String word;
        public long count;
        public WordWithCount() {
        }
        
        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return "WordWithCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }
    }
}
